package rddSummary.transition.key_value_type

import org.apache.spark.{SparkConf, SparkContext}

object test_aggregateByKey {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sparkContext = new SparkContext(conf)

    /**
     * 将数据根据不同的规则进行分区内计算和分区间计算
     */
    val dataRDD1 =
      sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
    val dataRDD2 =
      dataRDD1.aggregateByKey(0)(_+_,_+_)
    dataRDD2.collect().foreach(println)

//    // TODO : 取出每个分区内相同 key 的最大值然后分区间相加
//    // aggregateByKey 算子是函数柯里化，存在两个参数列表
//    // 1. 第一个参数列表中的参数表示初始值
//    // 2. 第二个参数列表中含有两个参数
//    //    2.1 第一个参数表示分区内的计算规则
//    //    2.2 第二个参数表示分区间的计算规则
//    val rdd =
//      sparkContext.makeRDD(List(
//        ("a",1),("a",2),("c",3),
//        ("b",4),("c",5),("c",6)
//      ),2)
//    // 0:("a",1),("a",2),("c",3) => (a,10)(c,10)
//    // 1:("b",4),("c",5),("c",6) => (b,10)(c,10)
//    val resultRDD =
//    rdd.aggregateByKey(10)(
//      (x, y) => math.max(x,y),
//      (x, y) => x + y
//    )
//    resultRDD.collect().foreach(println)

    sparkContext.stop()
  }

}
